#ifndef __M_CONSUMER_H__
#define __M_CONSUMER_H__

#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include <iostream>
#include <unordered_map>
#include <memory>
#include <vector>
#include <string>
#include <mutex>
#include <functional>

namespace wsmq
{
    using ConsumerCallback = std::function<void(const std::string&, const BasicProperties *, const std::string&)>;

    struct Consumer
    {
        using ptr = std::shared_ptr<Consumer>;
        std::string _tag;
        std::string _qname;
        bool _auto_ack;
        ConsumerCallback _callback;
        Consumer() 
        {
            DLOG("new a consumer: %p", this);
        }
        Consumer(const std::string &tag, const std::string &qname, bool auto_ack, ConsumerCallback callback) 
        : _tag(tag), _qname(qname), _auto_ack(auto_ack), _callback(callback)
        {
            DLOG("new a consumer: %p", this);
        }
        ~Consumer() 
        {
            DLOG("delete a consumer: %p", this);
        }
    };
    class QueueConsumer
    {
    public:
        using ptr = std::shared_ptr<QueueConsumer>;
        QueueConsumer(const std::string &qname) : _qname(qname), _rr_seq(0)
        {
        }
        Consumer::ptr insert(const std::string &tag, const std::string &qname,
                             bool auto_ack, ConsumerCallback callback)
        {
            // 加锁
            std::unique_lock<std::mutex> lock(_mutex);
            // 判断新增的消费者是否重复
            for (auto consumer : _consumers)
            {
                if (consumer->_tag == tag)
                {
                    DLOG("新增的消费者重复了");
                    return Consumer::ptr();
                }
            }
            // 不重复则构造对象
            auto consumer = std::make_shared<Consumer>(tag, qname, auto_ack, callback);
            // 添加管理后返回构造的对象
            _consumers.push_back(consumer);
            return consumer;
        }
        void remove(const std::string &tag)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto it = _consumers.begin(); it != _consumers.end(); it++)
            {
                if ((*it)->_tag == tag)
                {
                    _consumers.erase(it);
                    return;
                }
            }
            return;
        }
        Consumer::ptr choose() // RR轮转获取消费队列数据的消费者
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (_consumers.size() == 0)
                return Consumer::ptr();
            int index = _rr_seq % _consumers.size();
            _rr_seq++;
            return _consumers[index];
        }
        bool empty()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return (_consumers.size() == 0);
        }
        bool exist(const std::string &tag)
        {
            // std::unique_lock<std::mutex> lock(_mutex);
            for (auto &consumer : _consumers)
            {
                if (consumer->_tag == tag)
                {
                    return true;
                }
            }
            return false;
        }
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            std::vector<Consumer::ptr>().swap(_consumers);
            _rr_seq = 0;
        }

    private:
        std::string _qname;
        std::mutex _mutex;
        uint64_t _rr_seq; // 实现RR轮转的序号
        std::vector<Consumer::ptr> _consumers;
    };
    class ConsumerManager
    {
    public:
        using ptr = std::shared_ptr<ConsumerManager>;
        ConsumerManager()
        {
        }
        void initQueueConsumer(const std::string &qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _qconsumers.find(qname);
            if (it != _qconsumers.end())
            {
                return;
            }
            auto qconsumer = std::make_shared<QueueConsumer>(qname);
            _qconsumers.insert(std::make_pair(qname, qconsumer));
        }
        void destroyQueueConsumer(const std::string &qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _qconsumers.find(qname);
            if (it == _qconsumers.end())
            {
                return;
            }
            _qconsumers.erase(qname);
        }
        Consumer::ptr insert(const std::string &tag, const std::string &qname,
                             bool auto_ack, ConsumerCallback callback)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto qit = _qconsumers.find(qname);
                if (qit == _qconsumers.end())
                {
                    DLOG("未找到消费者队列 %s 的管理句柄，插入失败", qname.c_str());
                    return Consumer::ptr();
                }
                qcp = qit->second;
            }
            return qcp->insert(tag, qname, auto_ack, callback); // 自己有锁，线程安全，不用加锁
        }
        void remove(const std::string &tag, const std::string &qname)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto qit = _qconsumers.find(qname);
                if (qit == _qconsumers.end())
                {
                    DLOG("未找到消费者队列 %s 的管理句柄，移除消费者失败", qname.c_str());
                    return;
                }
                qcp = qit->second;
            }
            qcp->remove(tag);
        }
        Consumer::ptr choose(const std::string &qname)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto qit = _qconsumers.find(qname);
                if (qit == _qconsumers.end())
                {
                    DLOG("未找到消费者队列 %s 的管理句柄，获取消费者失败", qname.c_str());
                    return Consumer::ptr();
                }
                qcp = qit->second;
            }
            return qcp->choose();
        }
        bool exist(const std::string &tag, const std::string &qname)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto qit = _qconsumers.find(qname);
                if (qit == _qconsumers.end())
                {
                    DLOG("未找到消费者队列 %s 的管理句柄", qname.c_str());
                    return false;
                }
                qcp = qit->second;

            }
            return qcp->exist(tag);
        }
        bool empty(const std::string &qname)
        {
            QueueConsumer::ptr qcp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto qit = _qconsumers.find(qname);
                if (qit == _qconsumers.end())
                {
                    DLOG("未找到消费者队列 %s 的管理句柄", qname.c_str());
                    return false;
                }
                qcp = qit->second;

            }

            return qcp->empty();
        }
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _qconsumers.clear();
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, QueueConsumer::ptr> _qconsumers;
    };
}

#endif